package grobit

import grobit.wire.PeerConnection

import java.nio.ByteBuffer
import java.nio.channels.SelectionKey
import java.nio.channels.Selector
import java.nio.channels.SocketChannel
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicBoolean

class PeerEngine implements Runnable {
    private Selector selector
    private AtomicBoolean isShutdown = new AtomicBoolean(false)
    private ConcurrentHashMap<SelectionKey, PeerConnection> clients = [:]

    final static PeerEngine instance = new PeerEngine().startUp()

    PeerEngine startUp() {
        selector = Selector.open()
        new Thread(this).start()
        return this
    }

    void register(PeerConnection peerConnection) {
        SocketChannel channel = peerConnection.socketChannel
        channel.configureBlocking(false)
        SelectionKey key = channel.register(selector, SelectionKey.OP_READ)
        clients.put(key, peerConnection)
    }

    @Override
    void run() {
        while (!isShutdown.get()) {
            println 'PeerEngine: running...'

            while(selector.select(1000) == 0) {
//                println 'Waiting...'
            }
            println 'PeerEngine: selected...'
            Iterator iterator = selector.selectedKeys().iterator()
            while(iterator.hasNext()) {
                println 'PeerEngine: next...'
                SelectionKey selectionKey = iterator.next()
                iterator.remove()
                if(selectionKey.isReadable()) {
                    SocketChannel channel = selectionKey.channel() as SocketChannel
                    if(!channel.isOpen()) {
                        println 'Cancelling closed connection'
                        selectionKey.cancel()
                        clients.remove(selectionKey)
                        continue
                    }
                    ByteBuffer buffer = ByteBuffer.allocate(1024)
                    try {
                        if(channel.read(buffer) == -1) {
                            channel.close()
                            selectionKey.cancel()
                            clients.remove(selectionKey)
                            println 'Closed connection'
                            continue
                        }
                    } catch (Exception e) {
                        println 'Closed ' + clients.get(selectionKey).cId + ' on Exception: ' + e.getMessage()
                        try { channel.close() } catch (Exception x) { }
                        selectionKey.cancel()
                        continue
                    }
                    buffer.flip()
                    try {
                        println 'Read: ' + buffer.limit() + ' bytes.'
                        clients.get(selectionKey).accept(buffer)
                    } catch(Exception x) {
                        println 'Caught exception from client: '
                        x.printStackTrace()
                        selectionKey.cancel()
                        clients.remove(selectionKey)
                    }
                }
            }
        }
    }

    void shutdown() {
        isShutdown.set(true)
    }
}
